package kalp.web;

import org.apache.spark.storage.StorageLevel;
import org.apache.spark.streaming.StreamingContext;
import org.apache.spark.streaming.receiver.Receiver;
import org.apache.spark.streaming.twitter.TwitterInputDStream;

import scala.Option;
import twitter4j.Status;
import twitter4j.auth.Authorization;
import twitter4j.auth.OAuthAuthorization;
import twitter4j.conf.Configuration;
import twitter4j.conf.ConfigurationBuilder;

public class ExtendedTwitterStream extends TwitterInputDStream {

	public ExtendedTwitterStream(StreamingContext ssc) {
		super(ssc, new Option<Authorization>() {

			public int productArity() {
				// TODO Auto-generated method stub
				return 0;
			}

			public Object productElement(int arg0) {
				// TODO Auto-generated method stub
				return null;
			}

			public boolean canEqual(Object arg0) {
				// TODO Auto-generated method stub
				return false;
			}

			@Override
			public Authorization get() {
				// TODO Auto-generated method stub
				return null;
			}

			@Override
			public boolean isEmpty() {
				// TODO Auto-generated method stub
				return false;
			}
		}, null, StorageLevel.MEMORY_ONLY());
	}

	@Override
	public Receiver<Status> getReceiver() {
		Configuration conf = new ConfigurationBuilder()
				.setOAuthAccessToken(
						"2255068147-v32adeOq6b6wzM3KvnjusTHhuEykNxgkZ3PScyR")
				.setOAuthAccessTokenSecret(
						"vXkR9jtECj0CSPpH9lEyfsE8v8citcZ5N61VaqhKJll3n")
				.setOAuthConsumerKey("pcGABkK2IcOgImFtS20g")
				.setOAuthConsumerSecret(
						"v9KL2VMCvYslz5M7MLAu9DkVa3dcAEwNuSwOs5xU").build();
		Authorization auth = new OAuthAuthorization(conf);
		return new ExtendedTwReciever(auth, null, StorageLevel.MEMORY_ONLY());
	}

}
